package com.atguigu.agg;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**
 * @author yhm
 * @create 2024-04-02 10:58
 */
public class Test01_KeyBy {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        // 2. 读取数据源
        List<WaterSensor> waterSensors = Arrays.asList(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_2", 1000L, 40),
                new WaterSensor("sensor_2", 2000L, 21),
                new WaterSensor("sensor_3", 2000L, 20),
                new WaterSensor("sensor_3", 3000L, 32),
                new WaterSensor("sensor_4", 3000L, 40),
                new WaterSensor("sensor_4", 4000L, 41)
        );
        DataStreamSource<WaterSensor> dataStreamSource = env.fromCollection(waterSensors);

        // 3. 处理数据
        KeyedStream<WaterSensor, String> keyedStream = dataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.id + "_" + value.ts;
            }
        });

//        keyedStream.sum("vc")
//                .print();

        KeyedStream<WaterSensor, Tuple2<String, Long>> keyedStream1 = dataStreamSource.keyBy(new KeySelector<WaterSensor, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> getKey(WaterSensor value) throws Exception {
                return new Tuple2<>(value.id, value.ts);
            }
        });
//         javaBean使用maxBy只能写属性名
//        SingleOutputStreamOperator<WaterSensor> vc = keyedStream1.max("vc");
//        vc.print();
        KeyedStream<WaterSensor, String> keyedStream2 = dataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.id;
            }
        });

        keyedStream2.maxBy("vc").print();

        // 4. 输出


        // 5. 执行环境
        env.execute();
    }
}
